package com.feidee.data.report.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.feidee.data.report.constant.Constant;
import com.feidee.data.report.dao.IRealTimeReportDao;
import com.feidee.data.report.dao.IRedisDao;
import com.feidee.data.report.model.RealTimeMetricParam;
import com.feidee.data.report.model.RealTimeParam;
import com.feidee.data.report.model.RealTimeReportConfigModel.RealTimeConfigModel;
import com.feidee.data.report.model.RealTimeReportConfigModel.ViewSqlModel;
import com.feidee.data.report.service.IRealTimeReportServices;
import com.feidee.data.report.util.RealTimeReportThread;
import com.feidee.data.report.util.ToolUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.dom4j.Element;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Service("realTimeReportService")
public class RealTimeReportServicesImpl implements IRealTimeReportServices {

    public static Logger logger = Logger.getLogger(RealTimeReportServicesImpl.class);

    private static ExecutorService realTimeReportThreadPool = Executors.newFixedThreadPool(Constant
            .REAL_TIME_THREAD_POOL_SIZE);

    @Resource
    private IRealTimeReportDao realTimeReportDao;
    @Resource
    private IRedisDao redisDao;

    @Override
    public List<JSONObject> runSql(List<RealTimeMetricParam> views, Map<String, String> searchSql) {

        Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResults = new HashMap<>();
        try {
            List<RealTimeReportThread> realTimeReportTasks = new ArrayList<>();
            Set<Map.Entry<String, String>> entries = searchSql.entrySet();
            for (Map.Entry<String, String> entry : entries) {
                String key = entry.getKey();
                String redisKey = key.split("#", -1)[0];
                Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResult = redisDao
                        .queryRealTimeReportResult(redisKey);
                if (realTimeResult != null) {
                    String realTimeResultStr = realTimeResult.toString();
                    realTimeResultStr = realTimeResultStr.substring(0, realTimeResultStr.length() > 1000 ? 1000 :
                            realTimeResultStr.length());
                    logger.info("实时报表查询指标: " + key + " redis缓存命中，结果为: " + realTimeResultStr);
                    // 将各个SQL结果汇总到一起
                    addToResultMap(realTimeResult, realTimeResults);
                    continue;
                }
                logger.info("实时报表查询指标: " + key + " redis缓存无命中,开始查询数据库");
                realTimeReportTasks.add(new RealTimeReportThread(realTimeReportDao, redisDao, key, entry.getValue()));
            }
            List<Future<Map<Map<String, JSONObject>, Map<String, JSONObject>>>> futures = realTimeReportThreadPool
                    .invokeAll(realTimeReportTasks, Constant.REALTIME_WAIT_TIME, TimeUnit.SECONDS);
            for (Future<Map<Map<String, JSONObject>, Map<String, JSONObject>>> future : futures) {
                try {
                    Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResult = future.get();
                    // 某个查询失败跳过处理
                    if (realTimeResult == null) {
                        continue;
                    }
                    // 将各个SQL结果汇总到一起
                    addToResultMap(realTimeResult, realTimeResults);
                } catch (Exception e) {
                    logger.error("实时报表获取结果超时", e);
                    return null;
                }
            }
        } catch (Exception e) {
            logger.error("实时报表查询失败", e);
            return null;
        }
        // 转换结果为list
        List<JSONObject> realTimeResultsToList = transformToList(realTimeResults, views);
        return realTimeResultsToList;
    }

    // 根据查询参数获取查询SQL

    @Override
    public Map<String, String> getSearchSql(RealTimeParam realTimeParam) {
        // 存储结构： key：表名&维度&起始时间&结束时间&过滤条件#指标€引擎   value：SQL
        Map<String, String> searchSqlMap = new HashMap<>();
        try {

            List<RealTimeMetricParam> metricsParam = realTimeParam.getViews();
            String start_date = realTimeParam.getStart_date();
            String end_date = realTimeParam.getEnd_date();
            for (RealTimeMetricParam metric : metricsParam) {

                StringBuilder keyStrBuilder = new StringBuilder();
                String view_name = metric.getView_name();
                JSONObject dim_name = metric.getDim_name();
                List<String> metric_name = metric.getMetric_name();
                JSONObject metricFilter = metric.getFilter();
                keyStrBuilder.append(view_name).append("&");
                String filter = getFilterStr(metricFilter);
                String dimStr = "";
                String dimKeyStr = "";
                if (!dim_name.isEmpty()) {
                    Set<Map.Entry<String, Object>> entries = dim_name.entrySet();
                    for (Map.Entry<String, Object> dim_entry : entries) {
                        String key = dim_entry.getKey();
                        String value = (String) dim_entry.getValue();
                        dimStr += value + " AS " + key + ",";
                        dimKeyStr += value + ",";
                        keyStrBuilder.append(key).append(",");
                    }
                    dimKeyStr = dimKeyStr.substring(0, dimKeyStr.length() - 1);
                }
                keyStrBuilder.append("&").append(start_date).append("&").append(end_date).append("&")
                        .append(ToolUtils.md5(filter)).append("#");
                for (String metricStr : metric_name) {
                    keyStrBuilder.append(metricStr).append(",");
                }

                Map<String, String> viewInfo = realTimeReportDao.getView(view_name);
                String engine = viewInfo.get("engine");
                String keyStr = keyStrBuilder.append("€").append(engine).toString()
                        .replace(",&", "&")
                        .replace(",€", "€");
                String sql = viewInfo.get("sql");
                sql = sql.replace("#dim_select_str#", "".equals(dimStr) ? "" : dimStr)
                        .replace("#filter_str#", filter)
                        .replace("#dim_groupby_str#", (!sql.contains("#dim_groupby_str#") || "".equals(dimKeyStr)) ?
                                "" : "group by " + dimKeyStr)
                        .replace("#start_date#", start_date).replace("#end_date#", end_date);
                searchSqlMap.put(keyStr, sql);
//				logger.info("实时报表key为: " + keyList.toString() + "获取到的SQL为: " + sql);
            }
        } catch (Exception e) {
            logger.error("实时报表获取查询SQL失败", e);
            return null;
        }
        return searchSqlMap;
    }

    private String getFilterStr(JSONObject filter) {
        if (filter == null) {
            return "1=1";
        }
        StringBuilder filterBuilder = new StringBuilder();
        for (String dimKey : filter.keySet()) {
            List<String> dimValues = (List<String>) filter.get(dimKey);
            filterBuilder.append(dimKey).append(" in ('");
            for (String dimValue : dimValues) {
                filterBuilder.append(dimValue).append("','");
            }
            filterBuilder.append(") and ");
        }
        String filterStr = filterBuilder.toString().replace(",') and ", ") and ");
        if (!ToolUtils.isNotNull(filterStr)) {
            return "1=1";
        }
        return filterStr.substring(0, filterStr.length() - 5);
    }

    @Override
    public String getConfig() {
        String metas = realTimeReportDao.getConfMeta();
        return metas;
    }

    private void addToResultMap(Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResult,
                                Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResults) {
        try {
            if (realTimeResults.isEmpty()) {
                realTimeResults.putAll(realTimeResult);
            } else {
                Set<Map.Entry<Map<String, JSONObject>, Map<String, JSONObject>>> entries = realTimeResult.entrySet();
                for (Map.Entry<Map<String, JSONObject>, Map<String, JSONObject>> entry : entries) {
                    Map<String, JSONObject> key = entry.getKey();
                    Map<String, JSONObject> value = entry.getValue();
                    if (realTimeResults.containsKey(key)) {
                        realTimeResults.get(key).putAll(value);
                    } else {
                        realTimeResults.put(key, value);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("实时报表汇总结果失败!", e);
        }
    }

    private List<JSONObject> transformToList(Map<Map<String, JSONObject>, Map<String, JSONObject>> realTimeResults,
                                             List<RealTimeMetricParam> views) {
        List<JSONObject> realTimeResultsToList = null;
        try {
            Set<Map.Entry<Map<String, JSONObject>, Map<String, JSONObject>>> entries = realTimeResults.entrySet();
            realTimeResultsToList = new ArrayList<>();
            // 查询参数中需要获取的指标
            HashMap<String, List<String>> selectMetrics = new HashMap<>();
            for (int i = 0; i < views.size(); i++) {
                RealTimeMetricParam realTimeMetricParam = views.get(i);
                String view_name = realTimeMetricParam.getView_name();
                List<String> metric_name = realTimeMetricParam.getMetric_name();
                // 存放每个视图需要获取的指标名称
                selectMetrics.put(view_name, metric_name);
                // 同一次查询维度相同，只需记录一次(同一次查询必须是相同的维度)
                if (i == 0) {
                    JSONObject dim_name = realTimeMetricParam.getDim_name();
                    if (dim_name != null) {
                        Set<String> dims = dim_name.keySet();
                        List<String> dimsList = new ArrayList<>();
                        dimsList.addAll(dims);
                        // 存储本次查询的维度信息
                        selectMetrics.put("dims", dimsList);
                    }
                }
            }
            List<String> dimsList = selectMetrics.get("dims");
            for (Map.Entry<Map<String, JSONObject>, Map<String, JSONObject>> entry : entries) {
                Map<String, JSONObject> key = entry.getKey();
                Map<String, JSONObject> value = entry.getValue();
                JSONObject jsonObject = new JSONObject();
                // 取出需要的维度
                if (dimsList != null) {
                    JSONObject dimJsonObject = new JSONObject();
                    JSONObject dims = key.get("dims");
                    for (String dimsStr : dimsList) {
                        // snappyData获取的元数据均为大写，需要转换成大写结果
                        dimJsonObject.put(dimsStr, dims.get(dimsStr.toUpperCase()));
                    }
                    jsonObject.put("dims", dimJsonObject);
                }
                Set<Map.Entry<String, JSONObject>> valueEntries = value.entrySet();
                for (Map.Entry<String, JSONObject> valueEntry : valueEntries) {
                    String key1 = valueEntry.getKey();
                    JSONObject value1 = valueEntry.getValue();
                    // 取出view所对应的指标，只获取需要的指标
                    List<String> viewMetric = selectMetrics.get(key1);
                    JSONObject metricJsonObject = new JSONObject();
                    for (String metric : viewMetric) {
                        // snappyData获取的元数据均为大写，需要转换成大写结果
                        metricJsonObject.put(metric, value1.get(metric.toUpperCase()));
                    }
                    jsonObject.put(key1, metricJsonObject);
                }
                realTimeResultsToList.add(jsonObject);
            }
        } catch (Exception e) {
            logger.error("实时报表结果转换出错!", e);
        }
        return realTimeResultsToList;
    }

    @Override
    public String setConfig(RealTimeConfigModel realTimeConfigModel) {
        String code = "0";
        try {
            String businessCnName = realTimeConfigModel.getBusiness_cn_name();
            String businessEnName = realTimeConfigModel.getBusiness_en_name();
            String dimConfig = JSONObject.toJSONString(realTimeConfigModel.getDim_config());
            List<ViewSqlModel> viewSqlModels = realTimeConfigModel.getView_sql();

            String viewConfig = JSONObject.toJSONString(realTimeConfigModel.getView_config());
            String confMetaCode = realTimeReportDao.setConfMeta(businessCnName, businessEnName, dimConfig, viewConfig);
            String confViewCode = realTimeReportDao.setConfView(businessEnName, viewSqlModels);
            if (!"0".equals(confMetaCode) && !"0".equals(confViewCode)) {
                code = "-2";
                logger.info("实时报表配置元数据及视图设置均失败！");
            } else if (!"0".equals(confMetaCode)) {
                code = "-3";
                logger.info("实时报表配置元数据设置失败！");
            } else if (!"0".equals(confViewCode)) {
                code = "-4";
                logger.info("实时报表配置视图设置失败！");
            } else {
                logger.info("实时报表配置元数据及视图设置成功！");
            }
        } catch (Exception e) {
            code = "-5";
            logger.error("实时报表配置设置失败！", e);
        }
        return code;
    }

    /**
     * 实时报表聚合计算
     */
    @Override
    public JSONObject aggCal(RealTimeParam realTimeParam, Map<String, String> searchSqls) {

        List<JSONObject> sqlResult = runSql(realTimeParam.getViews(), searchSqls);
        JSONObject aggResult = new JSONObject();
        long startTime = System.currentTimeMillis();

        if (sqlResult == null) return null;
        if (sqlResult.size() < 1) return aggResult;
        try {

            /*
                {
                "dims": {
                    "loan_prd": "meijie_ss",
                     "dim_date": "2018-05-03"
                },
                "view_d": {
                    "total_user": "1",
                    "pass_user": "0",
                    "total_freq": "1"
                }
            }*/

            // view -> metric -> value
            Map<String, Map<String, BigDecimal>> max = new HashMap<>();
            Map<String, Map<String, BigDecimal>> min = new HashMap<>();
            Map<String, Map<String, BigDecimal>> avg = new HashMap<>();
            Map<String, Map<String, BigDecimal>> count = new HashMap<>();
            Map<String, Map<String, BigDecimal>> sum = new HashMap<>();

            for (JSONObject row : sqlResult) {
                for (String viewKey : row.keySet()) {
                    if ("dims".equals(viewKey)) continue;
                    Map<String, BigDecimal> maxViewResult = max.getOrDefault(viewKey, new HashMap<>());
                    Map<String, BigDecimal> minViewResult = min.getOrDefault(viewKey, new HashMap<>());
                    Map<String, BigDecimal> sumViewResult = sum.getOrDefault(viewKey, new HashMap<>());
                    Map<String, BigDecimal> countViewResult = count.getOrDefault(viewKey, new HashMap<>());

                    JSONObject viewMetrics = row.getJSONObject(viewKey);

                    for (String metricKey : viewMetrics.keySet()) {
                        String currentValStr = viewMetrics.getString(metricKey);
                        if (StringUtils.isEmpty(currentValStr)) {
                            logger.info("Empty metric " + metricKey + ", record info: " + row.toJSONString());
                            continue;
                        }
                        BigDecimal currentVal = new BigDecimal(currentValStr);

                        BigDecimal maxVal = maxViewResult.getOrDefault(metricKey, currentVal);
                        if (maxVal.compareTo(currentVal) < 0) maxVal = currentVal;
                        maxViewResult.put(metricKey, maxVal);

                        BigDecimal minVal = minViewResult.getOrDefault(metricKey, currentVal);
                        if (minVal.compareTo(currentVal) > 0) minVal = currentVal;
                        minViewResult.put(metricKey, minVal);

                        BigDecimal sumVal = sumViewResult.getOrDefault(metricKey, BigDecimal.ZERO);
                        sumViewResult.put(metricKey, sumVal.add(currentVal));

                        BigDecimal countVal = countViewResult.getOrDefault(metricKey, BigDecimal.ZERO);
                        countViewResult.put(metricKey, countVal.add(BigDecimal.ONE));
                    }

                    max.put(viewKey, maxViewResult);
                    min.put(viewKey, minViewResult);
                    sum.put(viewKey, sumViewResult);
                    count.put(viewKey, countViewResult);
                }
            }

            for (String viewKey : sum.keySet()) {
                Map<String, BigDecimal> metricsMap = new HashMap<>();
                for (String metricKey : sum.get(viewKey).keySet()) {
                    metricsMap.put(metricKey, sum.get(viewKey).get(metricKey).divide(count.get(viewKey).get
                            (metricKey), 4, RoundingMode.FLOOR));
                }
                avg.put(viewKey, metricsMap);
            }

            aggResult.put("max", max);
            aggResult.put("min", min);
            aggResult.put("avg", avg);
            aggResult.put("count", count);
            aggResult.put("sum", sum);
        } catch (Exception e) {
            logger.error("实时报表聚合计算异常", e);
            return null;
        }
        logger.info("实时报表聚合计算耗时" + (System.currentTimeMillis() - startTime) + "ms");
        return aggResult;
    }

    public static void main(String[] args) {
        RealTimeParam realTimeParam = JSONObject.parseObject("{\n" +
                "\"views\":\n" +
                "\t[{\n" +
                "\t\t\"view_name\":\"view_a\",\n" +
                "\t\t\"metric_name\":[\"bill_update\"]\n" +
                "\t},{\n" +
                "\t\t\"view_name\":\"view_b\",\n" +
                "        \"dim_name\":{\"dim_date\":\"bill_update_date\"},\n" +
                "\t\t\"metric_name\":[\"reg_import_bill\",\"reg_import_bill_mail\"]\n" +
                "\t}],\n" +
                "\"start_date\":\"2018-05-01\",\n" +
                "\"end_date\":\"2018-05-03\"\n" +
                "}", RealTimeParam.class);
        RealTimeReportServicesImpl realTimeReportServices = new RealTimeReportServicesImpl();
        Map<String, String> searchSql = realTimeReportServices.getSearchSql(realTimeParam);
        System.out.println(searchSql.toString());
        System.out.println(realTimeReportServices.getConfig());
    }
}
